package kafka.examples

import java.util.Properties
import java.util.concurrent.{ExecutionException, TimeUnit}

import org.apache.kafka.clients.producer._

/**
  * Created by hunter on 17-11-19.
  */
class Producer(private val topic: String, private val isAsync: Boolean) extends Thread {
  private var producer: KafkaProducer[Integer, String] = _

  val props: Properties = new Properties()
  props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, s"${KafkaProperties.KAFKA_SERVER_URL}:${KafkaProperties.KAFKA_SERVER_PORT}")
  props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer")
  props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer")
  props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
  props.put(ProducerConfig.ACKS_CONFIG,"1")
  this.producer = new KafkaProducer[Integer, String](props)


  override def run(): Unit = {
    var messageNo = 0
    while (true) {
      messageNo += 1
      TimeUnit.SECONDS.sleep(1)
      val messageStr = s"Message_${messageNo}"
      val startTime: Long = System.currentTimeMillis()
      if (isAsync) {
        producer.send(new ProducerRecord[Integer, String](
          topic,
          messageNo,
          messageStr),
          new DemoCallBack(startTime, messageNo, messageStr))
      } else {
        //send synchronously
        try{
          producer.send(new ProducerRecord[Integer,String](
            topic,
            messageNo,
            messageStr
          )).get()
        }catch{
          case e @ (_: InterruptedException | _: ExecutionException) => {
           e.printStackTrace()
          }
        }
      }
    }
  }
}


class DemoCallBack(private val startTime: Long,
                   private val key: Int,
                   private val message: String) extends Callback {
  override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
    val elapsedTime = System.currentTimeMillis - startTime
    if (metadata != null) {
      println(s"message(${key}, ${message}) sent to partition(${metadata.partition()}),offset(${metadata.offset()}) in ${elapsedTime} ms")
    } else {
      exception.printStackTrace()
    }
  }
}
